package cn.doitedu.day05

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.reflect.ClassTag

object T03_CustomSort {

  def main(args: Array[String]): Unit = {

    //1.创建SparkConf
    val conf = new SparkConf().setAppName("WordCount")
      .setMaster("local[4]") //如果提交到集群中运行，setMaster必须注释掉

    //2.创建SparkContext
    val sc = new SparkContext(conf)

    val data = List(
      "yuange,34,99.99",
      "jiege,30,99.99",
      "xingge,18,9999.99"
    )

    val rdd1 = sc.parallelize(data)

    val tfBoy: RDD[Boy] = rdd1.map(line => {
      val fields = line.split(",")
      val name = fields(0)
      val age = fields(1).toInt
      val fv = fields(2).toDouble
      new Boy(name, age, fv)
    })

    implicit val ord = new Ordering[Boy] {
      override def compare(x: Boy, y: Boy): Int = {
        if (x.fv == y.fv) {
          x.age - y.age
        } else {
          java.lang.Double.compare(y.fv, x.fv)
        }
      }
    }

    val sorted: RDD[Boy] = tfBoy.sortBy(boy => boy)

    println(sorted.collect().toBuffer)
  }

}

case class Boy(name: String, age: Int, fv: Double)

//class Boy(
//  val name: String,
//  var age: Int,
//  var fv: Double
//) extends Ordered[Boy] with Serializable {
//  override def compare(that: Boy): Int = {
//    if (this.fv == that.fv) {
//      this.age - that.age
//    } else {
//      java.lang.Double.compare(that.fv, this.fv)
//    }
//  }
//
//  override def toString = s"Boy($name, $age, $fv)"
//}
